package com.zkq.graduate.service.impl;

import com.alibaba.fastjson.JSON;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.rabbitmq.client.Channel;
import com.zkq.graduate.pojo.SendMessageResult;
import com.zkq.graduate.service.MqService;
import com.zkq.graduate.util.SendMessageUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.*;

/**
 * @Author zhangkaiqiang
 * @Date 2019/12/31  11:00
 * @Description TODO
 */
@Service("mqService")
@Slf4j
public class MqServiceImpl implements MqService {


	@Resource
	private RabbitTemplate rabbitTemplate;


	private final static String EXCHANGE_NAME = "graduate-message-exchange";


	@Override
	public void publish(Object params) {
		if(params!=null){
			String messageId= UUID.randomUUID().toString().trim().replaceAll("-","");
			CorrelationData  correlationData=new CorrelationData(messageId);
			rabbitTemplate.convertAndSend(EXCHANGE_NAME,"send.message",params,correlationData);
		}
	}


	/**
	 * 进行短信发送的方法
	 * @param channel
	 * @param message
	 */
	@RabbitListener(queues = "graduate-message")
	public void consumer(Message message, Channel channel){
		int corePoolSize=Runtime.getRuntime().availableProcessors()+1;
		ThreadFactory threadFactory=new ThreadFactoryBuilder().setNameFormat("message-consumer-%d").build();
		ExecutorService pool=new ThreadPoolExecutor(
				corePoolSize,
				10,
				1,
				TimeUnit.SECONDS,
				new LinkedBlockingDeque<>(30),
				threadFactory,
				new ThreadPoolExecutor.CallerRunsPolicy());
		String messageBody=new String(message.getBody());
		Map<String,Object> param= JSON.parseObject(messageBody,Map.class);;
		// 向线程池中提交任务--发送验证码
		String phone=String.valueOf(param.get("phone"));
		Integer captcha=(Integer)param.get("captcha");
		pool.submit(()->{
			try {
				String result= SendMessageUtil.sendSms(phone,"login",captcha);
				SendMessageResult sendMessageResult = JSON.parseObject(result, SendMessageResult.class);
				if(Objects.equals(sendMessageResult.getCode(),"OK")){
					log.info("成功向用户:"+phone+",发送手机验证码");
					channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
				}
			} catch (IOException e) {
				log.info("登录短信验证码发送失败,失败原因为:"+e.toString()+"进行重发操作");
				try {
					channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
				} catch (IOException e1) {
					e1.printStackTrace();
				}
			}
		});
	}
}
